/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl.disk;

import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import it.unimi.dsi.fastutil.longs.LongSet;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.sabayframework.mem.Transaction.Status;


/**
 * Disk transaction implementation.
 * Class is not thread-safe.
 * */
public class DiskTransactionImpl implements DiskTransaction {
		
	protected final Long2ObjectOpenHashMap<DataEntry> modifiedEntries;
	protected final Set<Integer> lockedLogs;
	protected final LongSet lockedEntries;
	protected final AbstractDiskStorage storage;
	
	protected long startedTime;
	protected long completedTime;
	protected Status status;
	private volatile boolean flushed;
	private IOException failure;
	
	protected DiskTransactionImpl(
			AbstractDiskStorage storage, 
			Long2ObjectOpenHashMap<DataEntry> modifiedEntries){
		this.startedTime = System.currentTimeMillis();
		this.storage = storage;
		this.modifiedEntries = modifiedEntries;
		this.flushed = true;
		this.lockedLogs = Collections.synchronizedSet(new HashSet<Integer>());  
		for(DataEntry e : modifiedEntries.values()){
			Integer key = e.getLogFile();
			lockedLogs.add(key);
		}
		lockedEntries = new LongOpenHashSet();
		status = Status.ACTIVE;
	}
	
	protected DiskTransactionImpl(AbstractDiskStorage storage) {
		this(storage,new Long2ObjectOpenHashMap<DataEntry>());
	}

	protected DataEntry[] getEntries(){
		DataEntry[] entries=modifiedEntries.values().toArray(new DataEntry[modifiedEntries.size()]);
		return entries;
	}
	/**
	 * Returns true if at least one address from given array is locked by
	 * this transaction.
	 * @param addresses array of addresses
	 * @return true if at least one address is locked by this transaction or false if none of the addresses 
	 * is locked by this transaction.
	 * */
	protected boolean isLocked(long[] addresses){
		if(!lockedEntries.isEmpty()){
			for(int i=0;i<addresses.length;++i)
				if(lockedEntries.contains(addresses[i])) return true;
		}
		return false;
	}
	protected void setLock(long[] addresses){
		for(int i=0;i<addresses.length;++i)	lockedEntries.add(addresses[i]);
	}
	
	public boolean tryLock(long[] addresses){
		boolean locked = true;
		for(int i=0;i<addresses.length;++i)
			if(!lockedEntries.contains(addresses[i])) {
				locked = false;
				break;
			}
		return locked? true : storage.tryLock(addresses, this) == this;
	}
	
	protected void checkActive() throws IOException {
		if(status!=Status.ACTIVE) throw new IOException("Transaction is not active"); 
	}
	
	public byte[] read(long address) throws IOException {
		DataEntry e = modifiedEntries.get(address);
		if(e==null) return storage.read(address);
		flush();
		return storage.read(e);
	}

	public void update(long address, byte[] data) throws IOException{
		checkActive(); 
		int length = data == null ? 0 : data.length;
		DataEntry e = new DataEntry(address, length, 0,0);
		if(length > 0){
			flushed = false;
			storage.write(e, data, 0);
			lockedLogs.add(e.getLogFile());
		}
		modifiedEntries.put(address, e);
	}
	
	public long getNextAvailableAddress() throws IOException {
		long[] address = new long[1];
		do{
			address[0] = storage.getNextAvailableAddress();
		} while(storage.tryLock(address, this)!=this);
		return address[0]; 
	}

	public long insert(byte[] data) throws IOException {
		long address = storage.getNextAvailableAddress();
		update(address, data);
		return address;
	}

	public void delete(long address) throws IOException {
		update(address, null);
	}

	public boolean hasModified(long address) throws IOException {
		return modifiedEntries.containsKey(address);
	}

	public long[] getModified() throws IOException {
		int len=modifiedEntries.size();
		long[] result = new long[len];
		Iterator<Long> it = modifiedEntries.keySet().iterator();
		int i=0;
		while(i < len && it.hasNext()){
			result[i++] = it.next();
		}
		return result;
	}
	
	public boolean hasLogLocked(int log){
		return lockedLogs.contains(log);
	}

	protected void complete() throws IOException{
		this.completedTime = System.currentTimeMillis();
		storage.completeDiskTransaction(this);
		modifiedEntries.clear();
		lockedEntries.clear();
		lockedLogs.clear();
	}
	public void commit() throws IOException{
		checkActive(); 
		status = Status.ROLLBACK;
		try{
			flush();
			checkFailure();
			DataEntry[] entries=getEntries();
			storage.commitEntries(entries);
			status = Status.COMMIT;
		} finally {
			complete();
		}
	}

	public void flush() throws IOException {
		if(!flushed){
			storage.flushCurrentLog();
			flushed = true;
		}
	}

	public void rollback() throws IOException {
		if(status != Status.ROLLBACK) {
			status = Status.ROLLBACK;
			complete();
		}
	}

	public boolean isCommitted() {
		return status == Status.COMMIT;
	}

	public boolean isActive() {
		return status == Status.ACTIVE;
	}

	protected boolean isFlushed(){
		return flushed;
	}
	protected synchronized IOException getFailure() {
		return failure;
	}
	
	protected synchronized void checkFailure() throws IOException {
		if(failure!=null) throw failure;
	}

	protected synchronized void setFailure(IOException failure) {
		this.failure = failure;
	}

	public long getActiveMillis() {
		return isActive() ? System.currentTimeMillis() - startedTime :
			completedTime - startedTime;
	}

	public Status getStatus() {
		return status;
	}
	
}
